/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson;

import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

import org.redisson.api.RFuture;
import org.redisson.api.RLock;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.RedisStrictCommand;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.pubsub.LockPubSub;

/**
 * 写锁: 排他锁、互斥锁
 */
public class RedissonWriteLock extends RedissonLock implements RLock {

    protected RedissonWriteLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
    }

    @Override
    String getChannelName() {
        return prefixName("redisson_rwlock", getRawName());
    }

    @Override
    protected String getLockName(long threadId) {
        return super.getLockName(threadId) + ":write";
    }

    /**
     * 写锁加锁方法
     */
    @Override
    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {

        /**
         * KEYS = Arrays.asList(getRawName())
         * KEYS[1] = getRawName()  即获取锁对象时设置的 “read_write_lock”
         *
         * ARGV = unit.toMillis(leaseTime), getLockName(threadId))
         * ARGV[1] = unit.toMillis(leaseTime)   默认30s
         * ARGV[2] = getLockName(threadId)      写锁名称，UUID:ThreadId:write
         *
         *
         * 写锁加锁两个逻辑：
         * 1、 如果当前没有线程持有锁，正常加锁，但可以返现写锁加锁流程和读锁不一样
         *     写锁的名称是UUID:ThreadId:write, 读锁是UUID:ThreadId
         *     写锁加锁不会单独创建一个记录当前锁过期时间的信息，
         *     这也是为什么在读锁重入中先获取写锁在获取读锁的情况下，只会有一个string类型的数据
         *
         * 2、 如果当前锁模式为write且获取锁的线程就是当前线程，可以正常重入
         *     read_write_lock里UUID:ThreadId:write的值 +1
         *     如果两个条件都不满足那么返回当前锁的过期时间，并自旋等待获取锁
         */
        return commandExecutor.syncedEval(getRawName(), LongCodec.INSTANCE, command,

                // local mode = hget redisson_read_write_lock mode
                "local mode = redis.call('hget', KEYS[1], 'mode'); " +

                // 无锁(无读锁,无写锁),  加读锁成功后 return null
                        // if (mode == false) then
                        //    hset redisson_read_write_lock mode write
                        //    hset redission_read_write_lock UUID:ThreadId:write 1
                        //    pexpire redisson_read_write_lock 30000
                        //    return null
                        // end
                        "if (mode == false) then " +
                        "redis.call('hset', KEYS[1], 'mode', 'write'); " +
                        "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return nil; " +
                        "end; " +

                // 锁降级, 读读兼容、读写互斥、写写互斥、写读互斥
                // 如果模式是write,这里可以隐蔽看出，如果模式是read,会等待所以的read完成后才能进入write模式
                // 1、只有写锁同时写锁的名称为当前线程
                        // if (mode == write && (hexists redisson_read_write_lock UUID:threadId:write == 1)) then
                        //     hincry redisson_read_write_lock UUID:threadId:write 1
                        //     long currentExpire = pttl redisson_read_write_lock
                        //     pexpire redisson_read_write_lock  currentExpire + 30000
                        //     return null
                        // end
                        "if (mode == 'write') then " +
                        // 获取锁线程是当前线程
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        // 锁重入次数 +1
                        // hincrby read_write_lock UUID:ThreadId:wwrite 1
                        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                        // 获取当前过期时间 并且重新设置
                        "local currentExpire = redis.call('pttl', KEYS[1]); " +
                        "redis.call('pexpire', KEYS[1], currentExpire + ARGV[1]); " +
                        "return nil; " +
                        "end; " +
                        "end;" +

                // 最后判断条件都不成立
                // 非当前线程的写锁
                // return pttl "redisson_read_write_lock"
                        "return redis.call('pttl', KEYS[1]);",
                Arrays.<Object>asList(getRawName()),
                unit.toMillis(leaseTime), getLockName(threadId));
    }


    /**
     * 写锁解锁方法
     */
    @Override
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {

        // 正确理解姿势
        // 1、没有写锁 或 当前线程全部解锁完          return 1
        // 2、当前线程还有重入锁(并没有完全解锁完成)   return 0
        // 3、读锁 或 其他线程持有锁                return null

        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,

                // local mode = hget redisson_read_write_lock mode
                "local mode = redis.call('hget', KEYS[1], 'mode'); " +

                // 无锁(无读锁,无写锁),发送订阅消息,结束
                // 无锁,return 1
                        // if (mode == false) then
                        //    PUBLISH redisson_rwlock:{redisson_read_write_lock} 0L
                        //    return 1
                        //    end

                        "if (mode == false) then " +
                        "redis.call(ARGV[4], KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end;" +

                // 只有mode为写锁,才能解锁, 否则不能进行解锁操作
                // 1、当前线程不持有锁,  return null
                // 2、当前线程全部解锁完  return 1
                // 3、当前线程还有重入锁(并没有完全解锁完成) return 0
                // 4、其他线程持有锁                return null

                        // if (mode == write) then
                        //    local lockExists = hexists redisson_read_write_lock UUID:threadId:write
                        //    if (lockExists == 0) then
                        //       return null
                        //    else
                        //       local counter = hincrby redisson_read_write_lock UUID:threadId:write -1
                        //       if (counter > 0) then
                        //          pexpire redisson_read_write_lock 30000
                        //          return 0;
                        //       else
                        //          hdel redisson_read_write_lock UUID:ThreadId:write
                        //          if (hlen redisson_read_write_lock == 1) then
                        //              del redisson_read_write_lock
                        //              PUBLISH redisson_rwlock:{redisson_read_write_lock} 1L
                        //          else
                        //              hset redisson_read_write_lock mode read
                        //          end
                        //          retrun 1
                        //       end
                        //    end
                        // end
                        "if (mode == 'write') then " +
                        "local lockExists = redis.call('hexists', KEYS[1], ARGV[3]); " +
                        "if (lockExists == 0) then " +
                        "return nil;" +
                        "else " +
                        "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                        "if (counter > 0) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                        "return 0; " +
                        "else " +
                        "redis.call('hdel', KEYS[1], ARGV[3]); " +
                        "if (redis.call('hlen', KEYS[1]) == 1) then " +
                        "redis.call('del', KEYS[1]); " +
                        "redis.call(ARGV[4], KEYS[2], ARGV[1]); " +
                        "else " +
                        "redis.call('hset', KEYS[1], 'mode', 'read'); " +
                        "end; " +
                        "return 1; " +
                        "end; " +
                        "end; " +
                        "end; "


                // 走到这里,说明 mode != write, 即不是写锁控制范围
                // mode是读锁    return null
                        + "return nil;",
                Arrays.<Object>asList(getRawName(), getChannelName()),
                LockPubSub.READ_UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId), getSubscribeService().getPublishCommand());
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException();
    }

    @Override
    protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
        CompletionStage<Boolean> f = super.renewExpirationAsync(threadId);
        return f.thenCompose(r -> {
            if (!r) {
                RedissonReadLock lock = new RedissonReadLock(commandExecutor, getRawName());
                return lock.renewExpirationAsync(threadId);
            }
            return CompletableFuture.completedFuture(r);
        });
    }

    @Override
    public RFuture<Boolean> forceUnlockAsync() {
        cancelExpirationRenewal(null);
        return commandExecutor.syncedEvalWithRetry(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('hget', KEYS[1], 'mode') == 'write') then " +
                        "redis.call('del', KEYS[1]); " +
                        "redis.call(ARGV[2], KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return 0; ",
                Arrays.asList(getRawName(), getChannelName()),
                LockPubSub.READ_UNLOCK_MESSAGE, getSubscribeService().getPublishCommand());
    }

    @Override
    public boolean isLocked() {
        RFuture<String> future = commandExecutor.writeAsync(getRawName(), StringCodec.INSTANCE, RedisCommands.HGET, getRawName(), "mode");
        String res = get(future);
        return "write".equals(res);
    }

}
